package com.edu.yx.app.App_04_DwsApp.Dws_03_Trade;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp_moreTopic;
import com.edu.yx.bean.TradeSourceOrderBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;

import static com.edu.yx.common.Constant.*;


public class Dws_01_Trade_Source_Order_Count extends BaseApp_moreTopic {


    public static void main(String[] args) {
        new Dws_01_Trade_Source_Order_Count().init(
                4031,
                2,
                "Dws_01_Trade_Source_Order_Count",
                TOPIC_DWD_TRADE_ORDER_INFO,
                TOPIC_DWD_TRAFFIC_UV_DETAIL
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {

        //1.转化成pojo，维度为来源，uv的时间戳用的系统时间戳
        DataStream<TradeSourceOrderBean> unionAll = unionAll(streams);

        //2.开窗聚合，用source分组
        SingleOutputStreamOperator<TradeSourceOrderBean> windowAndAgg = windowAndAgg(unionAll);

//        windowAndAgg.print();

        //3.写入doris
        writeDoris(windowAndAgg);

    }


    private void writeDoris(SingleOutputStreamOperator<TradeSourceOrderBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_trade_source_order_count"));

    }

    private SingleOutputStreamOperator<TradeSourceOrderBean> windowAndAgg(DataStream<TradeSourceOrderBean> unionAll) {

        SingleOutputStreamOperator<TradeSourceOrderBean> result = unionAll
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TradeSourceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getSource())
                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
                .sideOutputLateData(new OutputTag<TradeSourceOrderBean>("late") {
                })
                .reduce(new ReduceFunction<TradeSourceOrderBean>() {
                            @Override
                            public TradeSourceOrderBean reduce(TradeSourceOrderBean value1,
                                                               TradeSourceOrderBean value2) throws Exception {

                                value1.getOrderUserSet().addAll(value2.getOrderUserSet());

                                value1.setFinalAmount(value1.getFinalAmount() + value2.getFinalAmount());
                                value1.setOrderCt(value1.getOrderCt() + value2.getOrderCt());
                                value1.setUvCt(value1.getUvCt() + value2.getUvCt());


                                return value1;
                            }
                        }, new ProcessWindowFunction<TradeSourceOrderBean, TradeSourceOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<TradeSourceOrderBean> elements,
                                                Collector<TradeSourceOrderBean> out) throws Exception {

                                TradeSourceOrderBean bean = elements.iterator().next();
                                //先把集合的null全部去掉（union的时候，uv添加的null）
                                bean.getOrderUserSet().removeAll(Collections.singleton("null"));
                                bean.setOrderUserCt((long) bean.getOrderUserSet().size());

                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                out.collect(bean);

                            }
                        }
                );

//        result.print("main");
//        result.getSideOutput(new OutputTag<TradeSourceOrderBean>("late"){}).print("late");

        return result;

    }

    private DataStream<TradeSourceOrderBean> unionAll(HashMap<String, DataStreamSource<String>> streams) {


        //1.order的数据
        SingleOutputStreamOperator<TradeSourceOrderBean> sourceOrder = streams
                .get(TOPIC_DWD_TRADE_ORDER_INFO)
                .map(strjson -> {

                    JSONObject jsonObject = JSON.parseObject(strjson);

                    String userId = jsonObject.getString("user_id");
                    String source = jsonObject.getString("source");
                    Long finalAmount = jsonObject.getLong("final_amount");
                    Long ts = jsonObject.getLong("ts") * 1000L;

                    return new TradeSourceOrderBean(
                            "",
                            "",
                            "",
                            source,
                            0L,
                            new HashSet<>(Collections.singleton(userId)),
                            1L, //一条数就是一个订单
                            0L,
                            finalAmount,
                            ts
                    );

                });


        //2.order的数据
        SingleOutputStreamOperator<TradeSourceOrderBean> Uv = streams
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(strjson -> {

                    JSONObject jsonObject = JSON.parseObject(strjson);

                    String source = jsonObject.getJSONObject("common").getString("sc");
//                    Long ts = jsonObject.getLong("ts") ;
                    Long ts = System.currentTimeMillis() ;

                    return new TradeSourceOrderBean(
                            "",
                            "",
                            "",
                            source,
                            0L,
                            new HashSet<>(Collections.singleton("null")),
                            0L,
                            1L,  //来一个就是一个uv
                            0L,
                            ts
                    );

                });

        return sourceOrder.union(Uv);

    }


}
